package com.cloud.logger.zchain.core.zookeeper;

import com.cloud.logger.zchain.core.common.entity.AngelaLoggerProjectNode;
import com.cloud.logger.zchain.core.common.properties.ServerInfo;
import com.cloud.logger.zchain.core.constants.Constants;
import com.google.gson.Gson;
import com.google.gson.GsonBuilder;
import lombok.extern.slf4j.Slf4j;
import org.apache.curator.RetryPolicy;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.framework.recipes.cache.PathChildrenCache;
import org.apache.curator.framework.recipes.cache.PathChildrenCacheListener;
import org.apache.curator.framework.state.ConnectionState;
import org.apache.curator.retry.ExponentialBackoffRetry;
import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.KeeperException;
import org.apache.zookeeper.ZooDefs;
import org.apache.zookeeper.data.Stat;

import java.util.Collections;
import java.util.List;

@Slf4j
public class ZookeeperFactory {

    /**
     * zookeeper服务地址
     */
    private String hosts;
    /**
     * 會话的超时时间(毫秒)
     */
    private Integer sessionTimeOut;
    /**
     * 连接的超时时间(毫秒)
     */
    private Integer connectionTimeOut;
    /**
     * 命名空间
     **/
    private String nameSpace;
    /**
     * zookeeper管理对象
     */
    private CuratorFramework zkTools;
    /**
     * 独享队列节点
     */
    private String monopolyQueueNode;
    /**
     * 连接状态
     */
    private ConnectionState connectionState;
    /**
     * 会话ID
     */
    private long sessionId;

    private static Gson gson = new GsonBuilder().create();
    
    public String getHosts() {
        return hosts;
    }

    public void setHosts(String hosts) {
        this.hosts = hosts;
    }

    public Integer getSessionTimeOut() {
        return sessionTimeOut;
    }

    public void setSessionTimeOut(Integer sessionTimeOut) {
        this.sessionTimeOut = sessionTimeOut;
    }

    public Integer getConnectionTimeOut() {
        return connectionTimeOut;
    }

    public void setConnectionTimeOut(Integer connectionTimeOut) {
        this.connectionTimeOut = connectionTimeOut;
    }

    public String getNameSpace() {
        return nameSpace;
    }

    public void setNameSpace(String nameSpace) {
        this.nameSpace = nameSpace;
    }

    public String  getMonopolyQueueNode(){
        return this.monopolyQueueNode;
    }

    public CuratorFramework getZkTools(){
        return this.zkTools;
    }
    /**
     * Spring初始化方法
     */
    public void afterPropertiesSet() {

    }
    /**
     * Spring初始化方法
     */
    public void init(String host, String nameSpace, Integer connectionTimeOut) throws Exception {

        try {

            this.hosts = host;
            this.nameSpace = nameSpace;
            this.connectionTimeOut = connectionTimeOut;
            this.connection();
            this.addListener();
        } catch (Exception e) {
            throw new Exception("zookeeper init failed. ", e);
        }
    }

    private static class SingletonHolder {
        /**
         * 静态初始化器，由JVM来保证线程安全
         */
        private static ZookeeperFactory instance = new ZookeeperFactory();
    }

    public static ZookeeperFactory getInstance() {
        return SingletonHolder.instance;
    }

    /**
     * 连接
     */
    public void connection() {
        Constants.MONOPOLY = this.nameSpace;
        RetryPolicy retryPolicy = new ExponentialBackoffRetry(1000, Integer.MAX_VALUE);
        zkTools = CuratorFrameworkFactory.builder().connectString(hosts).retryPolicy(retryPolicy)
                .connectionTimeoutMs(connectionTimeOut == null ? 30000 : connectionTimeOut).sessionTimeoutMs(sessionTimeOut == null ? 30000 : sessionTimeOut).build();
        zkTools.start();
    }

    /**
     * 创建节点
     */
    public void createProjectNode(AngelaLoggerProjectNode node) throws Exception {

        String nodeStr = gson.toJson(node);

        String path = Constants.ZK_PROJECT_PATH
                +Constants.SEPARATOR
                +node.getCode()
                +Constants.RAIL
                +ServerInfo.ip
                +Constants.RAIL
                +ServerInfo.port;

        Stat stat =zkTools.checkExists().forPath(path);

        if(stat!=null)
            zkTools.delete().forPath(path);

        ZchainConnectionStateListener stateListener = new ZchainConnectionStateListener(path, nodeStr);
        zkTools.getConnectionStateListenable().addListener(stateListener);

        String nodeId =zkTools.create().
                creatingParentsIfNeeded().withMode(CreateMode.EPHEMERAL).
                forPath(path, nodeStr.getBytes());

        PathChildrenCache cache = new PathChildrenCache(zkTools, Constants.SEPARATOR+Constants.MONOPOLY, true);
        cache.start();

        PathChildrenCacheListener cacheListener = (zkTools, event) -> {
            switch (event.getType()) {
                case CHILD_ADDED: {
                    LeaderElection.getLeader();
                    break;
                }
                case CONNECTION_RECONNECTED: {
                    LeaderElection.getLeader();
                    break;
                }
                case CONNECTION_LOST: {
                    LeaderElection.getLeader();
                    break;
                }
                case CHILD_REMOVED: {
                    LeaderElection.getLeader();
                    break;
                }
            }
        };
        cache.getListenable().addListener(cacheListener);
        System.out.println(nodeId);
    }

    /**
     * 创建定时任务节点
     */
    public String createScheduleNode(String shceuldPath) throws Exception {

        String path =Constants.SEPARATOR
                +Constants.MONOPOLY
                +Constants.SEPARATOR
                +shceuldPath
                + Constants.SEPARATOR
                + Constants.QUEUE_NODE;

        Stat stat =zkTools.checkExists().forPath(path);

        if(stat!=null)
            zkTools.delete().forPath(path);
        return zkTools.create().
                withMode(CreateMode.EPHEMERAL_SEQUENTIAL).
                forPath(path);
    }

    /**
     * 创建zchain服务任务节点
     */
    public String createServerNode(String shceuldPath) throws Exception {

        String path =Constants.ZK_SERVER_PATH
                + Constants.SEPARATOR
                + Constants.QUEUE_NODE;

        Stat stat =zkTools.checkExists().forPath(path);

        if(stat!=null)
            zkTools.delete().forPath(path);
        return zkTools.create().
                withMode(CreateMode.EPHEMERAL_SEQUENTIAL).
                forPath(path,ServerInfo.ipPortByte);

    }

    /**
     * 连接状态监听
     */
    public void addListener() {
        zkTools.getConnectionStateListenable().addListener((curatorFramework, connectionState) -> {
            log.info("state: " + connectionState.name());
            this.connectionState = connectionState;
            if (ConnectionState.CONNECTED == connectionState) {
                try {
                    sessionId = zkTools.getZookeeperClient().getZooKeeper().getSessionId();
                    registerMonopolyQueue();
                } catch (Exception e) {
                    log.error("注册独占队列失败");
                }
            }
            if (ConnectionState.RECONNECTED == connectionState) {
                try {
                    if (sessionId != zkTools.getZookeeperClient().getZooKeeper().getSessionId()) {
                        registerMonopolyQueue();
                    }
                } catch (Exception e) {
                    log.error("注册独占队列失败");
                }
            }
            if (connectionState.equals(ConnectionState.LOST)) {
                System.out.println("lost");
            }
            if (connectionState.equals(ConnectionState.SUSPENDED)) {
            }
            if (connectionState.equals(ConnectionState.READ_ONLY)) {
            }
        });
    }

    /**
     * 注册独占队列
     */
    public void registerMonopolyQueue() throws Exception {
        if (zkTools.checkExists().watched().forPath(Constants.SEPARATOR + Constants.MONOPOLY) == null) {
            zkTools.create().creatingParentsIfNeeded().withMode(CreateMode.PERSISTENT).withACL(ZooDefs.Ids.OPEN_ACL_UNSAFE).forPath(Constants.SEPARATOR + Constants.MONOPOLY);
            log.info("创建独享锁队列节点成功！");
        }
        if (monopolyQueueNode == null || (monopolyQueueNode != null && zkTools.checkExists().forPath(monopolyQueueNode) == null)) {
            monopolyQueueNode = createServerEphemeralSequentialNode();
            log.info("成功加入独享锁队列");
            ServerInfo.setScheduleNode(monopolyQueueNode);
        }
    }

    public String createServerEphemeralSequentialNode() throws Exception {
        String node = zkTools.create().withMode(CreateMode.EPHEMERAL_SEQUENTIAL).withACL(ZooDefs.Ids.OPEN_ACL_UNSAFE)
                .forPath(Constants.SEPARATOR + Constants.MONOPOLY + Constants.SEPARATOR + Constants.QUEUE_NODE);
        ServerInfo.setScheduleNode(node);
        return node;
    }

    /**
     * 获得独占锁的执行权限
     * 
     * @return 执行权限标识
     * @throws KeeperException
     * @throws InterruptedException
     */
    public boolean getMonopolyLock() throws Exception {
        boolean flag = false;
        if (connectionState != null && (connectionState.equals("CONNECTED") || connectionState.equals("RECONNECTED"))) {
            List<String> nodes = zkTools.getChildren().watched().forPath(Constants.SEPARATOR + Constants.MONOPOLY);
            if (nodes.size() > 0) {
                Collections.sort(nodes);
                // 判断当前应用是否在队列的第一位
                if ((Constants.SEPARATOR + Constants.MONOPOLY + Constants.SEPARATOR + nodes.get(0)).equals(monopolyQueueNode)) {
                    flag = true;
                }
            }
        }
        return flag;
    }

    /**
     * 关闭连接
     */
    public void close() {
        if (zkTools != null) {
            zkTools.close();
            zkTools = null;
        }
    }
}
